﻿using ScheduleTask;
using System;
using System.Collections.Concurrent;
using System.Data;
using System.Threading;

namespace Qin.ScheduleTask.Common
{
    public abstract class HighSpeedTaskBase<T> : BaseTask where T : class
    {
        #region ctor
        public HighSpeedTaskBase(IServiceProvider serviceProvider, TaskType taskType, string taskName, int runPeriodInSecond)
            : this(serviceProvider, taskType, taskName, runPeriodInSecond, null, null)
        {
        }

        public HighSpeedTaskBase(IServiceProvider serviceProvider, TaskType taskType, string taskName, DateTime? workingStartTime = null)
            : this(serviceProvider, taskType, taskName, 0, workingStartTime, null)
        {
        }

        public HighSpeedTaskBase(IServiceProvider serviceProvider, TaskType taskType, string taskName, int runPeriodInSecond, DateTime? workingStartTime = null, DateTime? workingEndTime = null)
            : base(serviceProvider, taskType, taskName, runPeriodInSecond, workingStartTime, workingEndTime)
        {
            this._serviceStopCts = new CancellationTokenSource();
            //启动事件消费线程
            ThreadPool.QueueUserWorkItem(o => { StartEventHandler(); });//消费线程在程序启动的时候就启动，始终处于处理中或者待命的状态
        }
        #endregion

        /// <summary>
        /// 待处理事件队列
        /// </summary>
        private readonly ConcurrentQueue<T> _eventQueue = new ConcurrentQueue<T>();

        private readonly ManualResetEvent _eventConsumeLoopWaitResetEvent = new ManualResetEvent(false);

        private readonly Semaphore _eventConsumeThreadCreateSemaphore = new Semaphore(20, 20);//同时只能放5个任务

        /// <summary>
        /// 服务停止CTS
        /// </summary>
        private readonly CancellationTokenSource _serviceStopCts;

        protected void Enqueue(T model)
        {
            _eventQueue.Enqueue(model);
            _eventConsumeLoopWaitResetEvent.Set();//放信号，告诉等待程序已经有一个新的消息进队列了, 通知消费程序开始消费
        }

        /// <summary>
        /// 消息消费协调管理线程
        /// </summary>
        private void StartEventHandler()
        {
            while (true)
            {
                if (_serviceStopCts.IsCancellationRequested)
                {
                    Logger.Warn("事件消费协调管理线程 Receive Cancellation Request");
                    break;
                }

                _eventConsumeLoopWaitResetEvent.WaitOne(500);//原则上在没有事件入队列的时候，等待1s就再去尝试

                T queuedEvent = null;
                if (_eventQueue.TryDequeue(out queuedEvent))
                {
                    _eventConsumeThreadCreateSemaphore.WaitOne();

                    #region 消费当前从队列里面拿出来的事件
                    ThreadPool.QueueUserWorkItem(o =>
                    {
                        try
                        {
                            CosumeEvent(queuedEvent);
                        }
                        catch (Exception ex)
                        {
                            Logger.Warn(ex, $"{TaskName}发生了一个异常");
                        }
                        finally
                        {
                            _eventConsumeThreadCreateSemaphore.Release();
                        }
                    });//新开一个线程去处理当前的事件
                    #endregion
                }
                else
                {
                    _eventConsumeLoopWaitResetEvent.Reset();
                }
            }
        }
        protected abstract void CosumeEvent(T model);
        //private void TransScope(Func<object, object> p, IsolationLevel isolationLevel)
        //{
        //    throw new NotImplementedException();
        //}
    }
}
